##############################################################################
# 
# Zope Public License (ZPL) Version 1.0
# -------------------------------------
# 
# Copyright (c) Digital Creations.  All rights reserved.
# 
# This license has been certified as Open Source(tm).
# 
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
# 
# 1. Redistributions in source code must retain the above copyright
#    notice, this list of conditions, and the following disclaimer.
# 
# 2. Redistributions in binary form must reproduce the above copyright
#    notice, this list of conditions, and the following disclaimer in
#    the documentation and/or other materials provided with the
#    distribution.
# 
# 3. Digital Creations requests that attribution be given to Zope
#    in any manner possible. Zope includes a "Powered by Zope"
#    button that is installed by default. While it is not a license
#    violation to remove this button, it is requested that the
#    attribution remain. A significant investment has been put
#    into Zope, and this effort will continue if the Zope community
#    continues to grow. This is one way to assure that growth.
# 
# 4. All advertising materials and documentation mentioning
#    features derived from or use of this software must display
#    the following acknowledgement:
# 
#      "This product includes software developed by Digital Creations
#      for use in the Z Object Publishing Environment
#      (http://www.zope.org/)."
# 
#    In the event that the product being advertised includes an
#    intact Zope distribution (with copyright and license included)
#    then this clause is waived.
# 
# 5. Names associated with Zope or Digital Creations must not be used to
#    endorse or promote products derived from this software without
#    prior written permission from Digital Creations.
# 
# 6. Modified redistributions of any form whatsoever must retain
#    the following acknowledgment:
# 
#      "This product includes software developed by Digital Creations
#      for use in the Z Object Publishing Environment
#      (http://www.zope.org/)."
# 
#    Intact (re-)distributions of any official Zope release do not
#    require an external acknowledgement.
# 
# 7. Modifications are encouraged but must be packaged separately as
#    patches to official Zope releases.  Distributions that do not
#    clearly separate the patches from the original work must be clearly
#    labeled as unofficial distributions.  Modifications which do not
#    carry the name Zope may be packaged in any form, as long as they
#    conform to all of the clauses above.
# 
# 
# Disclaimer
# 
#   THIS SOFTWARE IS PROVIDED BY DIGITAL CREATIONS ``AS IS'' AND ANY
#   EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
#   IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
#   PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL DIGITAL CREATIONS OR ITS
#   CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
#   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
#   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
#   USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
#   ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
#   OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
#   OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
#   SUCH DAMAGE.
# 
# 
# This software consists of contributions made by Digital Creations and
# many individuals on behalf of Digital Creations.  Specific
# attributions are listed in the accompanying credits file.
# 
##############################################################################
"""
DCOracle ZODB Storage

$Id: Full.py,v 1.6 2002/02/08 18:14:27 matt Exp $
"""

import ZODB.BaseStorage, ZODB.POSException, time, DCOracle2

from ZODB.utils import U64, u64, p64
from string import join, find, atol, strip
from ZODB.TimeStamp import TimeStamp
from DCOracle2 import dbiRaw
from binascii import b2a_base64, a2b_base64

class DCOracleStorageError(ZODB.POSException.StorageError):
    "A DCOracleStorage-specific exception occurred"

class DCOracleStorageSystemError(
    DCOracleStorageError, ZODB.POSException.StorageSystemError):
    "A DCOracleStorage-specific internal exception occurred"

# We really need to rethink the close method for DCOracle cursors.
# In the mean time, we'll provide this lamo close method that
# just closes at the oracle level without destroying the cursor.

class DCOracleStorage(ZODB.BaseStorage.BaseStorage):

    # Note, we have lots of attributes that are database objects like
    # stored procedures, prepared statements, and cursors. They all have
    # names begining with "_sql_".

    def __init__(self, connector, name='Oracle Storage', prefix='zodb_'):
        ZODB.BaseStorage.BaseStorage.__init__(self, name)
        if type(connector) is type(''):
            connector=lambda c=connector: DCOracle2.connect(c)
        self._connector=connector
        self._prefix=prefix
        self._connect()

    def _connect(self):
        try: self._sql_connection.close()
        except: pass
        connection=self._connector()
        prefix=self._prefix
        self._sql_connection = connection
        self._getProcedures(connection, prefix)
        self._sql_insertData = connection.prepare(
            "insert into %spickle values (:z_oid, :z_tid, :z_pickle)"
            % prefix)
        self._sql_pickle_cursor = connection.cursor()

        try: oid=self._sql_max_oid()
        except: pass
        else:
            if oid: self._oid=pack64(oid)

        try: tid=self._sql_max_tid()
        except: pass
        else:
            if tid:
                self._ts=TimeStamp(pack64(tid))

        self._sql_history_cursor=None
        self._sql_versions_cursor=None
        self._sql_undo_cursor=None
        self._sql_oids_cursor=None
        self._sql_oids_cursor=None

    def _getProcedures(self, connection, prefix):
        zodb_methods = getattr(connection.procedures, prefix+'methods')
        for p in (
            'abortVersion', 'begin', 'commitVersion', 'history',
            'len', 'load', 'size', 'load_serial', 'max_oid',
            'store', 'undo', 'undoLog',
            'version', 'versions', 'versionEmpty',
            ):
            setattr(self, '_sql_'+p, getattr(zodb_methods, prefix+p))


    def __len__(self):
        self._lock_acquire()
        try: return self._call('len')
        finally: self._lock_release()

    def _getColumn(self, c, index=0, f=(lambda x: x)):
        result=[]; a=result.append
        while 1:
            r=c.fetchone()
            if not r: break
            a(f(r[index]))

        c.close()
        return result

    def _call(self, name, args=(), cursor=None):
        name='_sql_'+name
        try:
            if cursor:
                cursor='_sql_'+cursor+'_cursor'
                c=getattr(self, cursor)
                if c is None:
                    c=self._sql_connection.cursor()
                    setattr(self, cursor, c)
                #return apply(getattr(self, name), args+(c,)), c
                return apply(getattr(self, name), args+(c,))
            else:
                return apply(getattr(self, name), args)
        except:
            if self._transaction is None:
                self._connect()
                if cursor:
                    c=getattr(self, cursor)
                    if c is None:
                        c=self._sql_connection.cursor()
                        setattr(self, cursor, c)
                    #return apply(getattr(self, name), args+(c,)), c
                    return apply(getattr(self, name), args+(c,))
                else:
                    return apply(getattr(self, name), args)
            else:
                raise

    def abortVersion(self, src, transaction):
        
        if transaction is not self._transaction:
            raise POSException.StorageTransactionError(self, transaction)
        
        self._lock_acquire()
        try:
            x, c = self._call('abortVersion', (self._z_serial, src), 'oids')
            return self._getColumn(c, 0, pack64)
        finally: self._lock_release()
        
    def close(self):
        self._lock_acquire()
        try:
            try: self._sql_connection.close()
            except: pass
        finally: self._lock_release()

    def commitVersion(self, src, dest, transaction, abort=None):
        # We are going to commit by simply storing back pointers.

        if src==dest: return
        
        if transaction is not self._transaction:
            raise POSException.StorageTransactionError(self, transaction)
        
        self._lock_acquire()
        try:
            x, c = self._call('commitVersion', (self._z_serial, src, dest),
                              'oids')
            return self._getColumn(c, 0, pack64)
        finally: self._lock_release()

    def getSize(self):
        self._lock_acquire()
        try: return ("%s records" % self._call('size'))
        finally: self._lock_release()


    def load(self, oid, version, _stuff=None):
        self._lock_acquire()
        try:
            z_oid = unpack64(oid)
            s, c = self._call('load', (z_oid, version), 'pickle')
            if not s:
                raise KeyError, oid
            p=c.fetchone()
            c.close()
            if not p:
                raise KeyError, oid
            p=p[0]
            return p, pack64(s)
        finally: self._lock_release()

    def loadSerial(self, oid, serial):
        self._lock_acquire()
        try:
            z_oid=unpack64(oid)
            z_serial=unpack64(serial)
            c = self._call('load_serial', (z_oid, z_serial), 'pickle')
            p=c.fetchone()
            c.close()
            if not p: raise KeyError, serial
            return p[0]
        finally: self._lock_release()
                    
    def modifiedInVersion(self, oid):
        self._lock_acquire()
        try: return self._call('version', (unpack64(oid),))
        finally: self._lock_release()

    def store(self, oid, serial, data, version, transaction):
        if transaction is not self._transaction:
            raise POSException.StorageTransactionError(self, transaction)

        self._lock_acquire()
        try:
            newserial=self._serial
            z_oid=unpack64(oid)
            z_serial=serial and unpack64(serial) or '0'
            z_newserial=self._z_serial
            if len(data) < 32000: # max length of long raw in PL/SQL
                r=self._sql_store(z_oid, z_serial, version, z_newserial,
                                  dbiRaw(data))
                data='' # mark as consumed so we don't do insert below.
            else:
                r=self._sql_store(z_oid, z_serial, version, z_newserial, None)
                
            if r:
                if r[0]=='c':
                    print "Conflict: oid '%s', old tid %s, new tid %s" % (
                        z_oid, z_serial, z_newserial)
                    raise ZODB.POSException.ConflictError, (
                        serial, pack64(r[1:]))
                elif r[0]=='v':
                    raise ZODB.POSException.VersionLockError, (
                        `oid`, r[1:])
                else:
                    raise DCOracleStorageSystemError, r

            if data:
                self._sql_insertData.execute(
                    z_oid = z_oid,
                    z_tid = z_newserial,
                    z_pickle = dbiRaw(data)
                    )
            
            return newserial
        
        finally: self._lock_release()

    def supportsUndo(self): return 1
    def supportsVersions(self): return 1

    def tpc_begin(self, transaction, tid=None, status=' '):
        self._lock_acquire()
        try:
            if self._transaction is transaction: return
            self._lock_release()
            self._commit_lock_acquire()
            self._lock_acquire()

            user=transaction.user
            desc=transaction.description
            ext=transaction._extension
            if ext:
                ext=encode_str(dumps(ext,1))
            else: ext=""

            if tid is None:
                t=time.time()
                t=apply(TimeStamp,(time.gmtime(t)[:5]+(t%60,)))
                self._ts=t=t.laterThan(self._ts)
                s=`t`
            else:
                self._ts=TimeStamp(tid)
                s=tid
                
            self._serial=s
            s=unpack64(s)
            self._z_serial=s
            self._call('begin',(s, status, user, desc, ext))
            self._transaction=transaction
            
        finally: self._lock_release()

    def tpc_abort(self, transaction):
        self._lock_acquire()
        try:
            if transaction is not self._transaction: return
            self._sql_connection.rollback()
            self._transaction=None
            self._commit_lock_release()
        finally: self._lock_release()

    def tpc_finish(self, transaction, f=None):
        self._lock_acquire()
        try:
            if transaction is not self._transaction: return
            if f is not None: f()
            self._sql_connection.commit()
        finally:
            self._transaction=None
            self._commit_lock_release()
            self._lock_release()

    def undo(self, transaction_id):
        self._lock_acquire()
        try:
            x, c = self._call('undo', (transaction_id,), 'oids')
            return self._getColumn(c, 0, pack64)
        finally: self._lock_release()

    def undoLog(self, first, last, filter=None):
        self._lock_acquire()
        try:
            c = self._call('undoLog', ())
            
            i=0; r=[]; a=r.append
            while i < last:
                row=c.fetchone()
                if not row: break
                tid, user, desc, ext = row
                d={'id': tid,
                   'time': TimeStamp(pack64(tid)).timeTime(),
                   'user_name': user or '',
                   'description': desc or '',
                   }
                if ext:
                    try: 
                        ext=loads(decode_str(ext))
                        d.update(ext)
                    except: pass
                if filter is None or filter(d):
                    if i >= first: a(d)
                    i=i+1

            c.close()

            return r
        finally: self._lock_release()

    def versionEmpty(self, version):
        self._lock_acquire()
        try: return self._call('versionEmpty', (version,))
        finally: self._lock_release()

    def versions(self, max=None):
        self._lock_acquire()
        try:
            c = self._call('versions', (), 'versions')
            return self._getColumn(c, 0)
        finally: self._lock_release()

    def history(self, oid, version=None, length=1, filter=None):
        self._lock_acquire()
        try:
            c = self._call('history', (unpack64(oid),), 'history')
            r=[]; a=r.append; i=0
            while len(r) < length:
                row=c.fetchone()
                if not row: break
                tid, user, desc, ext, ver = row
                ver=ver or ''
                if version is not None and ver != version: continue
                serial=pack64(tid)
                d={'time': TimeStamp(serial).timeTime(),
                   'user_name': user,
                   'description': desc,
                   'serial': serial,
                   'version': ver,
                   'size': 0,
                   }
                if ext:
                    try: 
                        ext=loads(decode_str(ext))
                        d.update(ext)
                    except: pass
                
                if filter is None or filter(d):
                    r.append(d)

            c.close()
            return r
        finally: self._lock_release()

    _packing=None
    def pack(self, t, referencesf):

        self._lock_acquire()
        try:
            if self._packing:
                raise DCOracleStorageError, 'Already packing'
            self._packing=1
        finally: self._lock_release()
            
        try:
        
            stop=`apply(TimeStamp, time.gmtime(t)[:5]+(t%60,))`
            if stop=='\0\0\0\0\0\0\0\0':
                raise DCOracleStorageError, 'Invalid pack time'
            tid=unpack64(stop)

            # Note that we open a special connection for packing
            connection=self._connector()
            zodb_methods = prefix_handler(
                getattr(connection.procedures,
                        self._prefix+'methods'),
                self._prefix)
            
            commit = connection.commit

            ################################################################
            # Begin packaing by marking transaction records
            # and making sure we have something to pack.
            if not zodb_methods.packMark(tid):
                return # there's nothing to be done!
            commit()

            ################################################################
            ## Figure out the current records as of the pack time. 
            ## Later, we will remove the reachable records 
            zodb_methods.packGetCurrent(tid)
            commit()

            ################################################################
            ## Remove the non-current data 
            zodb_methods.packRemoveNonCurrent(tid)
            commit()

            ################################################################
            ## Filter reachable objects from packCurrent in prepration for
            ## record GC
            c=connection.prepare(
                """select z_pickle 
                from %spickle p, %srecord r
                where r.z_oid = :i_oid and r.z_tid < :i_tid and
                p.z_oid = r.z_oid and
                p.z_tid = r.z_tid"""
                % (self._prefix, self._prefix)
                )
            c2=connection.prepare(
                "delete from %spackCurrent where z_oid = :i_oid"
                % self._prefix)
            packPickles=zodb_methods.packPickles
            rootl = ['\0\0\0\0\0\0\0\0']; pop = rootl.pop
            i=0
            # Holger Cornelsen contributed a duplicate OID patch
            pindex = {}                                     # HC
            referenced = pindex.has_key                     # HC
            while rootl:        
                i=i+1
                poid = pop()                                # HC
                if referenced(poid):                        # HC
                    continue                                # HC
                pindex[poid] = 1                            # HC
                oid=unpack64(poid)                          # HC
                __traceback_info__=rootl, i, oid, tid
                #packPickles(unpack64(oid), tid, c)
                c2.execute(i_oid=oid)
                c.execute(i_oid=oid, i_tid=tid)
                while 1:
                    row=c.fetchone()
                    if not row:
                        #oclose(c)
                        break
                    data=row[0]
                    referencesf(data, rootl)
                commit()

            ################################################################
            ## GC unreachable current records as of the pack time */
            zodb_methods.packGC(tid)
            commit()

            ################################################################
            ## GC no-longer referenced pickle records 
            zodb_methods.packGCPickles(tid)
            commit()

            ################################################################
            ## GC no-longer referenced transaction records 
            zodb_methods.packGCTrans(tid)
            commit()

        finally:
            # clean-up
            self._packing=None

    def iterator(self): return DBIterator(self._connector, self._prefix)

class prefix_handler:
    # lamo prefix mapper
    def __init__(self, methods, prefix):
        self.__methods=methods
        self.__prefix=prefix

    def __getattr__(self, name):
        r=getattr(self.__methods, self.__prefix+name)
        setattr(self, name, r)
        return r

class Iterator:
    """A General simple iterator that uses the Python for-loop index protocol
    """
    __index=-1
    __current=None

    def __getitem__(self, i):
        __index=self.__index
        while i > __index:
            __index=__index+1
            self.__current=self.next(__index)

        self.__index=__index
        return self.__current


class DBIterator(Iterator):
    """Iterate over the transactions in a DCOracleStorage database.
    """
    
    def __init__(self, connectionf, prefix='zodb_'):
        self._sql_connection=connectionf()
        self._tc=self._sql_connection.cursor()
        self._tc.execute(
            """\
            select to_char(z_tid), z_status, z_user, z_desc, z_ext
            from %strans order by z_tid asc
            """ % prefix)
        self._rc=self._sql_connection.prepare(
            """
            select to_char(r.z_oid), to_char(r.z_tid), z_version, z_pickle
            from %srecord r, %sversions v, %spickle p
            where r.z_tid = :i_tid and v.z_vid = r.z_vid
                  and p.z_oid = r.z_oid and p.z_tid = r.z_data
            """ % (prefix, prefix, prefix))

    def next(self, index=0):
        r=self._tc.fetchone()
        if not r: raise IndexError, index
        return RecordIterator(self._rc, r)

class RecordIterator(Iterator):
    """Iterate over the records in a DCOracleStorage transaction.
    """

    def __init__(self, statement, record):
        self._statement=statement
        tid, self.status, user, description, ext = record
        self.tid=pack64(tid)
        self._extension=ext and loads(decode_str(ext)) or {}
        self.user=user or ''
        self.description=description or ''
        statement.execute(i_tid=tid)

    def next(self, index):
        r=self._statement.fetchone()
        if not r: raise IndexError, index
        return Record(r)

class Record:

    def __init__(self, record):
        oid, tid, version, p = record
        self.oid=pack64(oid)
        self.serial=pack64(tid)
        self.version=version or '' # guard against None
        self.data=p
    
        

def encode_str(state):
    "Convert a string to an encoded string"
    l=len(state)

    if l > 57:
        states=[]; append=states.append
        for i in range(0,l,57):
            append(b2a_base64(state[i:i+57])[:-1])
        state=join(states,'')
    else: state=b2a_base64(state)[:-1]

    l=find(state,'=')
    if l >= 0: state=state[:l]
        
    return state

def decode_str(state):
    "Convert an encoded string to a sequence"
    l=len(state)

    if l > 76:
        states=[]; append=states.append
        j=0
        for i in range(l/76):
            k=j+76
            append(a2b_base64(state[j:k]))
            j=k

        if j < l:
            state=state[j:]
            l=len(state)
            k=l%4
            if k: state=state+'='*(4-k)
            states.append(a2b_base64(state))
        state=join(states,'')
    else:
        l=len(state)
        k=l%4
        if k: state=state+'='*(4-k)
        state=a2b_base64(state)

    return state


def pack64(v, st=type('')):
    if type(v) is st: return p64(atol(v))
    return p64(v)

def unpack64(v):
    return U64(v)
